package org.jgroups.tests;

import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.logging.Log;
import org.jgroups.logging.LogFactory;
import org.jgroups.protocols.FragHeader3;
import org.jgroups.protocols.FRAG3;
//import org.apache.commons.logging.Log;
//import org.apache.commons.logging.LogFactory;

import java.io.File;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.List;
import java.util.ArrayList;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: Mar 17, 2009
 * Time: 11:41:59 AM
 * To change this template use File | Settings | File Templates.
 */
public class TestSender  extends ReceiverAdapter {
    private static Log logger = LogFactory.getLog(TestSender.class);

    private List<Long> begin = new ArrayList<Long>();
    private List<Long> current;
    AtomicInteger no ;
    int count;
    FRAG3 frag3;


    public TestSender(int count) {
        this.count = count ;
        this.no = new AtomicInteger(0);
        List<String> list = runShell(new String[] { "sh","-c","netstat -s | grep -A4 Udp" }) ;
        for ( String each : list) {
            String[] strs = each.split("\\s");
            begin.add( Long.valueOf( strs[0]) );
        }
        current = begin;
        logger.info( output(begin));
    }

    public void receive(Message msg) {
        if ( msg == null ) {
            logger.warn("Mesasge is null ");
        }
        else {
            FragHeader3 header = (FragHeader3) msg.getHeader("FRAG3") ;
            logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ msg.getLength()
                    +" destination="+ msg.getDest()+" "+ ( header == null ? "null" : header.toString()));
        }
        int i = no.getAndIncrement();
        if ( i % count == (count -1 ) ) {
            List<Long> now =  new ArrayList<Long>();
            logger.info("FRAG3 " + frag3.toString() );
            List<String> list = runShell(new String[] { "sh","-c","netstat -s | grep -A4 Udp" }) ;
            for ( String each : list) {
                String[] strs = each.split("\\s");
                now.add( Long.valueOf( strs[0]) );
            }
            if ( now.size() > 2 ) {
                long dif = now.get(2) - current.get(2);
                if ( dif == 0 ) logger.info(i+" messages no udp error");
                else logger.info(i+" messages extra udp error " + dif);
                current = now;
            }
        }
    }


    public static void help() {
          System.out.println("java stress.org.jgroups.test.TestSender -udp ./udp.xml -cluster test-1 -count 100 size -size 10");
    }


    public static String output(List<Long> list) {
        StringBuffer sb = new StringBuffer();
        for (int i=0 ; i < list.size() ; i++) {
           switch (i) {
           case 0 : sb.append("packets received " + list.get(i).toString());
               break;

           case 1 : sb.append(" unknown port received " + list.get(i).toString());
               break;
           case 2 : sb.append(" packet receive errors " + list.get(i).toString());
               break;
           case 3 : sb.append(" packets sent " + list.get(i).toString());
               break;
           default :
               logger.warn("i "+ i +" is greater than 4");
           }
        }
        return sb.toString();
    }

    public static String ltrim(String str) {
        StringBuffer sb = new StringBuffer( str.length());
        boolean on = true ;
        for ( int i=0; i < str.length() ; i++) {
            if ( ! (on && str.charAt(i) == ' ' ) ) {
                sb.append( str.charAt( i));
                on = false;
            }
        }
        return sb.toString();
    }

    /**
     *
     * @param cmdline
     * @return list of result string
     */
    public static List<String> runShell(String[] cmdline) {
        List<String> list = new ArrayList<String>();
        try {
            Process proc = Runtime.getRuntime().exec( cmdline);
            BufferedReader br = new BufferedReader( new InputStreamReader( proc.getInputStream()));
            String line = null;
            int i = 0;
            while ( (line = br.readLine()) != null) {
                //skip the first line
                if ( i > 0 ) list.add( ltrim(line));
                i++;
                if( i > 4 ) break;
            }
            return list;
        } catch (Exception ex ) {
            System.out.println( ex.getMessage());
            return list;
        }
    }


    public static void main(String[] args) throws Exception {
        String[] defaults = { "./udp.xml","test-1" , "100" , "20", "2000", "3" };

        for ( int i=0; i < args.length ; ) {
            if ( args[i].equals("-udp")) defaults[0] = args[++i] ;
            else if ( args[i].equals("-cluster")) defaults[1] = args[++i] ;
            else if ( args[i].equals("-count")) defaults[2] = args[++i] ;
            else if ( args[i].equals("-size")) defaults[3] = args[++i] ;
            else if ( args[i].equals("-times")) defaults[4] = args[++i] ;
            else if ( args[i].equals("-sleep")) defaults[5] = args[++i] ;
            else if ( args[i].equals("-help")) help() ;
            else {
               help();
            }
            i ++;
        }
        int c =  Integer.valueOf( defaults[2]);
        TestSender ts = new TestSender( c);
        System.setProperty("java.net.preferIPv4Stack", "true");
        logger.info("java.net.preferIPv4Stack ="+ System.getProperty("java.net.preferIPv4Stack"));

        JChannel channel = new JChannel(new File(defaults[0]));
        channel.setReceiver(ts);
        channel.connect(defaults[1]);
        int count = Integer.valueOf( defaults[2]);
        int size  = Integer.valueOf( defaults[3]);
        int times =  Integer.valueOf( defaults[4]);
        int sleep = Integer.valueOf( defaults[5]);
//        Address dest = new IpAddress("10.129.120.156", 9500);
        FRAG3 frag3 = (FRAG3) channel.getProtocolStack().getTopProtocol();
        ts.frag3 = frag3;
        for (int j= 0; j < times ; j ++ ) {
            logger.info("Begin iteration " + j );
            long begin = System.nanoTime();
            for (int i = 0;  i < count ; i ++) {
                int s = ( i % size ) +1 ;
                Message msg = new Message(null, null, new byte[60000* s] );
                channel.send( msg);
            }
            long end = System.nanoTime();
            logger.info("Iteration "+ j +"  duration micro seconds "+TimeUnit.MICROSECONDS.convert((end - begin), TimeUnit.NANOSECONDS ));
            logger.info("End iteration " + j +" and sleep for "+ sleep+"  seconds" );
            Thread.sleep( 1000L * sleep );
            logger.info("FRAG3 "+ frag3.toString());
        }

    }


}
